package com.example.sourcecodelearning.config;

import com.example.framework.mq.rocketmq.config.RocketMQProperties;
import com.example.framework.mq.rocketmq.core.service.RocketMQConsumerService;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableConfigurationProperties(RocketMQProperties.class)
public class RocketMQConfiguration {

    @Resource
    private RocketMQConsumerService<?> rocketMQService;

    /**
     * 注册mq消费者并配置监听类
     *
     * @return mq消费者
     * @throws MQClientException 异常
     */
    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public DefaultMQPushConsumer defaultMQPushConsumer(RocketMQProperties rocketMQProperties) throws MQClientException {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMQProperties.getConsumerGroup());
        // 设置NameServer的地址
        consumer.setNamesrvAddr(rocketMQProperties.getNamesrvAddr());
        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe(rocketMQProperties.getTopic(), "*");
        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(rocketMQService);
        return consumer;
    }

}
